package com.ztesoft.web.backup;



import java.math.BigDecimal;
import java.math.RoundingMode;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.Callable;

import javax.sql.DataSource;

import org.springframework.jdbc.datasource.DataSourceUtils;
import org.springframework.jdbc.support.JdbcUtils;

import com.ztesoft.core.db.ds.DynamicDataSourceManager;
import com.ztesoft.core.idproduce.ISequenceGenerator;
import com.ztesoft.core.spring.context.SpringApplicationContextHolder;
import com.ztesoft.framework.exception.BaseAppException;
import com.ztesoft.framework.log.ZTEsoftLogManager;
import com.ztesoft.framework.util.DateUtils;
import com.ztesoft.framework.util.Utils;
import com.ztesoft.web.baseconfig.db.arg.DmTableRelArg;
import com.ztesoft.web.baseconfig.db.arg.DmTargetTableArg;
import com.ztesoft.web.baseconfig.db.arg.SubTableDefineArg;
import com.ztesoft.web.baseconfig.db.arg.TargetColMapArg;
import com.ztesoft.web.baseconfig.db.dao.DmSrcTableDao;
import com.ztesoft.web.baseconfig.db.dao.DmTableRelDao;
import com.ztesoft.web.baseconfig.db.dao.DmTargetTableDao;
import com.ztesoft.web.baseconfig.db.dao.SubTableDefineDao;
import com.ztesoft.web.baseconfig.db.dao.TargetColMapDao;
import com.ztesoft.web.baseconfig.db.po.DmSrcTablePO;
import com.ztesoft.web.baseconfig.db.po.DmTableRelPO;
import com.ztesoft.web.baseconfig.db.po.DmTargetTablePO;
import com.ztesoft.web.baseconfig.db.po.SubTableDefinePO;
import com.ztesoft.web.baseconfig.db.po.TargetColMapPO;
import com.ztesoft.web.cleanup.service.CleanUpService;
import com.ztesoft.web.common.DMSConstant;
import com.ztesoft.web.common.customdb.dao.DmSrcTableCustomDao;
import com.ztesoft.web.common.db.arg.DmArg;
import com.ztesoft.web.common.db.arg.DmSecondaryStageArg;
import com.ztesoft.web.common.db.dao.DmDao;
import com.ztesoft.web.common.db.dao.DmSecondaryStageDao;
import com.ztesoft.web.common.db.po.DmPO;
import com.ztesoft.web.common.db.po.DmSecondaryStagePO;
import com.ztesoft.web.common.exchange.ExchangeFunc;
import com.ztesoft.web.common.vo.TaskNotify;
import com.ztesoft.web.resourcecheck.ResourceCheckJob;
import com.ztesoft.web.syncbaseinfo.service.BaseInfoService;
import com.ztesoft.web.taskgenerator.db.arg.DmTaskStageArg;
import com.ztesoft.web.taskgenerator.db.dao.DmTaskStageDao;
import com.ztesoft.web.taskgenerator.db.po.DmTaskStagePO;

/**
 * 备份处理线程
 * Created by Chen.Shitian on 2015/8/1.
 */
public class BackUpThread implements Callable<String>{
	private static final ZTEsoftLogManager LOGGER = ZTEsoftLogManager.getLogger(BackUpThread.class);
	private ThreadLocal<Boolean> Flag = new ThreadLocal<Boolean>(){
		@Override
		protected Boolean initialValue() {
			return true;
		}
	};
	private ThreadLocal<String> errorMsg = new ThreadLocal<String>();
	private ThreadLocal<List<DmSrcTablePO>> allBackUpTable = new ThreadLocal<List<DmSrcTablePO>>();//记录已备份的分表
	private DmTaskStagePO dtsPO;
	private Integer buCount;
    private ThreadLocal<DmTaskStagePO> taskStagePO = new ThreadLocal<DmTaskStagePO>();
    private ThreadLocal<DmTaskStagePO> clearPO = new ThreadLocal<DmTaskStagePO>();
    private ThreadLocal<Boolean> continueFlag = new ThreadLocal<Boolean>(){
    	@Override
		protected Boolean initialValue() {
			return true;
		}
    };//继续备份标识
    private ThreadLocal<Integer> backUpCount = new ThreadLocal<Integer>();//单次备份数量
    private ThreadLocal<DmSrcTablePO> srcTablePO = new ThreadLocal<DmSrcTablePO>();//源对象
    private ThreadLocal<DmTargetTablePO> targetTablePO = new ThreadLocal<DmTargetTablePO>();//目标对象
    private ThreadLocal<List<DmPO>> dmList = new ThreadLocal<List<DmPO>>();
    private ThreadLocal<List<Long>> keyList = new ThreadLocal<List<Long>>();
    //定义backUpMap保存备份的明细标识，key为目标表表名，value为明细标识列表
	private ThreadLocal<Map<String,List<Long>>> backUpMap = new ThreadLocal<Map<String,List<Long>>>();
	//定义childSourceNameMap保存从表对应的实际表名，key为从表标识，value为实际表名Map{表名，条数}
	private ThreadLocal<Map<Integer,Map<String,Integer>>> childSourceNameMap = new ThreadLocal<Map<Integer,Map<String,Integer>>>();
	//定义childRecordSizeMap保存从表对应的单条记录大小，key为从表标识，value为记录大小
	private ThreadLocal<Map<Integer,Integer>> childRecordSizeMap = new ThreadLocal<Map<Integer,Integer>>();
	//定义sourceDataTypeMap保存主表备份字段的数据库字段类型，key为字段名，value为数据库字段类型
	private ThreadLocal<Map<String,Integer>>sourceDataTypeMap = new ThreadLocal<Map<String,Integer>>();
	//定义childSourceDataTypeMap保存从表备份字段的数据库字段类型，key为字段名，value为数据库字段类型
	private ThreadLocal<Map<String,Integer>>childSourceDataTypeMap = new ThreadLocal<Map<String,Integer>>();
	private ThreadLocal<List<DmSrcTablePO>>childSourceList = new ThreadLocal<List<DmSrcTablePO>>();//从表列表
	//将查询结果放入sourceResultMap中。Long为detailId，String为字段名，Object为字段值
	private ThreadLocal<Map<Long, Map<String, Object>>> sourceResultMap = new ThreadLocal<Map<Long,Map<String,Object>>>();
	//将查询结果放入sourceResultMap中。Long为detailId，String为字段名，Object为字段值
	private ThreadLocal<Map<Long, List<Map<String, Object>>>> childSourceResultMap = new ThreadLocal<Map<Long,List<Map<String,Object>>>>();
	private ThreadLocal<String> taskStageState = new ThreadLocal<String>();
	
	private ISequenceGenerator sequenceGenerator;//主键生成器
    private DmTaskStageDao dmTaskStageDao;//任务环节
    private DmSrcTableDao srcTableDao;//源对象
    private DmTargetTableDao targetTableDao;//目标对象
    private DmDao dmDao;//体检结果明细
    private TargetColMapDao targetColMapDao;//目标字段映射
	private SubTableDefineDao subTableDefineDao;//分表定义
	private ExchangeFunc exchangeFunc;//转换方法
	private DmSrcTableCustomDao srcTableCustomDao;
	private DmTableRelDao dmTableRelDao;//关联表
	private DmSecondaryStageDao secondaryStageDao;//从表环节
	private DynamicDataSourceManager dynamicDataSourceManager;//数据库 连接
	private CleanUpService cleanUpService;//清理接口
    private BaseInfoService baseInfoService;
    private ResourceCheckJob resourceCheckJob;
	
    public BackUpThread(DmTaskStagePO taskStagepo,int backCount){
    	LOGGER.debug("|BackUpThread初始化线程.任务Id：{0},策略Id:{1}",taskStagepo.getDmTaskId(),taskStagepo.getDmStrategyId());
    	dtsPO =taskStagepo;
    	buCount = backCount;
    	resourceCheckJob = SpringApplicationContextHolder.getBean(ResourceCheckJob.class);
    	sequenceGenerator = (ISequenceGenerator) SpringApplicationContextHolder.getBean("sequenceProcGenerator");
    	dmTaskStageDao =  SpringApplicationContextHolder.getBean(DmTaskStageDao.class);
    	srcTableDao = SpringApplicationContextHolder.getBean(DmSrcTableDao.class);
    	targetTableDao = SpringApplicationContextHolder.getBean(DmTargetTableDao.class);
    	dmDao = SpringApplicationContextHolder.getBean(DmDao.class);
    	targetColMapDao = SpringApplicationContextHolder.getBean(TargetColMapDao.class);
    	subTableDefineDao = SpringApplicationContextHolder.getBean(SubTableDefineDao.class);
    	exchangeFunc = SpringApplicationContextHolder.getBean(ExchangeFunc.class);
    	srcTableCustomDao = SpringApplicationContextHolder.getBean(DmSrcTableCustomDao.class);
    	dmTableRelDao = SpringApplicationContextHolder.getBean(DmTableRelDao.class);
    	secondaryStageDao =  SpringApplicationContextHolder.getBean(DmSecondaryStageDao.class);
    	dynamicDataSourceManager =  SpringApplicationContextHolder.getBean(DynamicDataSourceManager.class);
    	baseInfoService = SpringApplicationContextHolder.getBean(BaseInfoService.class);
    	cleanUpService = SpringApplicationContextHolder.getBean(CleanUpService.class);
    }
    
    private void init(){
    	 taskStagePO.set(dtsPO.cloneBO());
    	 backUpCount.set(buCount.intValue());
    }
 
	@Override
	public String call()  {
		try{
			init();
			LOGGER.info(Thread.currentThread()+"任务调度|++===备份服务|任务ID={0},策略ID={1}BEGIN..table[{2}]",
					taskStagePO.get().getDmTaskId(),taskStagePO.get().getDmStrategyId(),taskStagePO.get().getCheckObjName());
			return BackUpTask();
		}catch(Exception e){
			LOGGER.error("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"错误信息:",e);
			taskStagePO.get().setState(DMSConstant.Status.FALSE.toString());
			taskStagePO.get().setErrorDesc(e.toString());
			taskStagePO.get().setStateDate(DateUtils.getNowDate());
			clearPO.get().setState(DMSConstant.Status.FALSE.toString());
			clearPO.get().setErrorDesc("备份失败");
			clearPO.get().setStateDate(DateUtils.getNowDate());
			dmTaskStageDao.updateByPrimaryKeySelective(taskStagePO.get());
			dmTaskStageDao.updateByPrimaryKeySelective(clearPO.get());
			return DMSConstant.Status.FALSE.toString();
		}
	}

    private String BackUpTask() throws Exception{
    	LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"获取清理环节记录");
    	DmTaskStageArg taskStageArg = new DmTaskStageArg();
		DmTaskStageArg.DmTaskStageCriteria c = taskStageArg.createCriteria();
		c.andDmTaskIdEqualTo(taskStagePO.get().getDmTaskId());
		c.andDmStrategyIdEqualTo(taskStagePO.get().getDmStrategyId());
		c.andStageEqualTo(DMSConstant.STAGE_CLEANUP);
		c.andCheckObjNameEqualTo(taskStagePO.get().getCheckObjName());//biaoming
		c.andStateEqualTo(DMSConstant.Status.PENDING.toString());
		List<DmTaskStagePO> taskStageList = dmTaskStageDao.selectByArg(taskStageArg);
		if(taskStageList.size()>0){
			clearPO.set(taskStageList.get(0));
		}else{
			Thread.sleep(10*1000);
			LOGGER.info("找不到待清理环节表，休眠10s再度查询~,还查询不到则返回失败");
			taskStageList = dmTaskStageDao.selectByArg(taskStageArg);
			if(taskStageList.size()>0){
				clearPO.set(taskStageList.get(0));
			} else {
				taskStagePO.get().setState(DMSConstant.Status.FALSE.toString());
				taskStagePO.get().setErrorDesc("找不到清理的任务环节表");
				taskStagePO.get().setStateDate(DateUtils.getNowDate());
				dmTaskStageDao.updateByPrimaryKeySelective(taskStagePO.get());
//			dmTaskStageDao.updateByPrimaryKeySelective(clearPO.get());
				return DMSConstant.Status.FALSE.toString();
			}
		}
    	LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"修改任务环节表的状态，状态时间，开始时间");
        //修改任务环节表的状态，状态时间，开始时间，
    	LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"dmTaskStageDao========"+dmTaskStageDao.toString());
        taskStagePO.get().setState(DMSConstant.Status.PROCESSING.toString());
        taskStagePO.get().setStateDate(DateUtils.getNowDate());
        if(Utils.isEmpty(taskStagePO.get().getTaskStartTime())){
        	taskStagePO.get().setTaskStartTime(DateUtils.getNowDate());
        }
        dmTaskStageDao.updateByPrimaryKeySelective(taskStagePO.get());
    	LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"在源数据表对象表（dm_src_table）中查询出归档方式");
    	//在源数据表对象表（dm_src_table）中查询出归档方式
        srcTablePO.set(srcTableDao.selectByPrimaryKey(taskStagePO.get().getSourceDataId()));
		//查找目标对象
		LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"查找目标对象");
		DmTargetTableArg dmTargetTableArg = new DmTargetTableArg();
		DmTargetTableArg.DmTargetTableCriteria targetTableCriteria = dmTargetTableArg.createCriteria();
		targetTableCriteria.andSourceDataIdEqualTo(srcTablePO.get().getSourceDataId());
		targetTablePO.set(targetTableDao.selectByArg(dmTargetTableArg).get(0));
		
        TaskNotify notify;
        String code;
        String msg;
        String cleanUpState = null;
        while (continueFlag.get()){
        	notify =  DMSConstant.taskInstList.get(taskStagePO.get().getDmTaskId());
            code = notify.getStatus();//异常编码
            msg = notify.getDealDesc();//异常信息
        	
        	 //判断是否接收到异常信号
        	if(DMSConstant.Status.PAUSE.toString().equals(code)||DMSConstant.Status.SYSPAUSE.toString().equals(code)){
                // 若为“00P”(暂停)则更新任务环节表（dm_task_stage）的状态，状态时间，异常原因，语句如下①；
        		LOGGER.info(Thread.currentThread()+"任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"暂停处理{0}",code);
        		taskStagePO.get().setState(DMSConstant.Status.PENDING.toString());
        	    taskStagePO.get().setStateDate(DateUtils.getNowDate());
        	    taskStagePO.get().setErrorDesc(msg);
        	    dmTaskStageDao.updateByPrimaryKey(taskStagePO.get());
        	    taskStageState.set(code);
        		continueFlag.set(false);
        	}else if(DMSConstant.Status.CANCEL.toString().equals(code)){//
        		/* 
        		 * 
        		 * 在备份环节取消的话，是需要将已备份的数据从目标数据库中清理掉，未备份的数据明细记录删除，明细记录表的数据都要删除。
        		 */
                // 若为“00C”（取消）则先更新任务环节表（dm_task_stage）的状态，状态时间，异常描述，
        		taskStageState.set(code);
        		taskStagePO.get().setState(DMSConstant.Status.CANCELLED.toString());
        	    taskStagePO.get().setStateDate(DateUtils.getNowDate());
        	    taskStagePO.get().setErrorDesc(msg);
        	    dmTaskStageDao.updateByPrimaryKey(taskStagePO.get());
                // 再根据任务体检结果（dm_*）的关键字段对目标表数据进行删除，语句如下②，最后根据任务标识删除任务体检结果（dm_*）
        		LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"取消处理");
        	    DmArg arg = new DmArg();
    		    DmArg.DmCriteria criteria = arg.createCriteria();
    		    criteria.andDmTaskIdEqualTo(taskStagePO.get().getDmTaskId());
    		    criteria.andDmStrategyIdEqualTo(taskStagePO.get().getDmStrategyId());
     		    criteria.andStateEqualTo(DMSConstant.Status.TOBACKUP.toString());
    		    dmDao.deleteByArg("DM_"+taskStagePO.get().getCheckObjName(), arg);
    		    //更新清理的任务环节表
				DmTaskStageArg stageArg = new DmTaskStageArg();
				DmTaskStageArg.DmTaskStageCriteria stageC = stageArg.createCriteria();
				stageC.andDmTaskIdEqualTo(taskStagePO.get().getDmTaskId());
				stageC.andDmStrategyIdEqualTo(taskStagePO.get().getDmStrategyId());
				stageC.andStageEqualTo(DMSConstant.STAGE_CLEANUP);
				DmTaskStagePO stagePo = new DmTaskStagePO();
				stagePo.setState(DMSConstant.Status.CANCELLED.toString());
				stagePo.setErrorDesc("备份取消");
				stagePo.setStateDate(DateUtils.getNowDate());
//    		    //调用清理 
				LOGGER.error("backup cancelled to call cleanup task 任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"调用清理 ");
				cleanUpState = cleanUpService.cleanUp(taskStagePO.get().getDmTaskId(), taskStagePO.get().getDmStrategyId(), DMSConstant.STAGE_BACKUP);
				if (cleanUpState != null) {				
					LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"更新清理的任务环节表");	
					dmTaskStageDao.updateByArgSelective(stagePo, stageArg);
				}
				dmTaskStageDao.updateByArgSelective(stagePo, stageArg);
        		continueFlag.set(false);
        	}else{
    			//变量重置
        		LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"变量重置");
        		taskStageState.set("");
    		 	dmList.set(new ArrayList<DmPO>());
    		 	keyList.set(new ArrayList<Long>());
    			backUpMap.set(new HashMap<String, List<Long>>());
    			childSourceNameMap.set(new HashMap<Integer, Map<String,Integer>>());
    			childRecordSizeMap.set(new HashMap<Integer, Integer>());
    			sourceDataTypeMap.set(new HashMap<String, Integer>());
    			childSourceDataTypeMap.set(new HashMap<String, Integer>());
    			childSourceList.set(new ArrayList<DmSrcTablePO>());//从表列表
    			sourceResultMap.set(new HashMap<Long, Map<String, Object>>());
    			childSourceResultMap.set(new HashMap<Long, List<Map<String,Object>>>());
    			allBackUpTable.set(new ArrayList<DmSrcTablePO>());
        		LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"|提取任务体检结果表（dm_*）的记录");
        		//根据配置文件的单次备份数量（默认1000）提取任务体检结果表（dm_*）的记录保存在列表中
        		//Select * from dm_xxx where dm_task_id=? And dm_strategy_id=? And state=’00B’order by create_date limit 1000;
        		DmArg dmArg = new DmArg();
        		DmArg.DmCriteria dmCriteria = dmArg.createCriteria();
        		dmCriteria.andDmTaskIdEqualTo(taskStagePO.get().getDmTaskId());
        		dmCriteria.andDmStrategyIdEqualTo(taskStagePO.get().getDmStrategyId());
        		dmCriteria.andStateEqualTo(DMSConstant.Status.TOBACKUP.toString());
        		dmArg.setOrderByClause("create_date");
        		dmList.set(dmDao.selectByArgAndSize("DM_"+taskStagePO.get().getCheckObjName(),dmArg,backUpCount.get()));
        		LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"获取体检明细记录"+dmList.get().size());
        		if(dmList.get().size()<=0){
            		LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"明细执行完成");
        			//更新任务环节表（dm_task_stage）的状态，状态时间，任务结束时间
            		LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"更新任务环节表");
        			taskStagePO.set(dmTaskStageDao.selectByPrimaryKey(taskStagePO.get().getDmInfoId()));
        			taskStagePO.get().setState(DMSConstant.Status.SUCCESS.toString());
        			taskStagePO.get().setStateDate(DateUtils.getNowDate());
        			taskStagePO.get().setTaskEndTime(DateUtils.getNowDate());
        			dmTaskStageDao.updateByPrimaryKey(taskStagePO.get());
        			//更新从表的任务环节表的状态，状态时间，任务结束时间
        			if (DMSConstant.ARCHIVE_TYPE_REL==srcTablePO.get().getArchiveType()) {
                		LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"更新从表任务环节表");
        				DmSecondaryStageArg secondaryStageArg = new DmSecondaryStageArg();
        				DmSecondaryStageArg.DmSecondaryStageCriteria secondaryStageCriteria = secondaryStageArg.createCriteria();
        				secondaryStageCriteria.andDmInfoIdEqualTo(taskStagePO.get().getDmInfoId());
        				secondaryStageCriteria.andStateEqualTo(DMSConstant.Status.PROCESSING.toString());
        				secondaryStageCriteria.andDmTaskIdEqualTo(taskStagePO.get().getDmTaskId());
        				secondaryStageCriteria.andStageEqualTo(DMSConstant.STAGE_BACKUP);
        				secondaryStageCriteria.andDmStrategyIdEqualTo(taskStagePO.get().getDmStrategyId());
        				DmSecondaryStagePO secondaryStagePO = new DmSecondaryStagePO();
        				secondaryStagePO.setState(DMSConstant.Status.SUCCESS.toString());
        				secondaryStagePO.setStateDate(DateUtils.getNowDate());
        				secondaryStagePO.setTaskEndTime(DateUtils.getNowDate());
        				secondaryStageDao.updateByArgSelective(secondaryStagePO, secondaryStageArg);
        			}
        			continueFlag.set(false);
        		}else{
            		int archiveMode = srcTablePO.get().getArchiveMode(); 
        			if (archiveMode == DMSConstant.ARCHIVE_MODE_CONDITION) {
        				//1：按条件归档；
        				try{
        					keyList.set(getKeyList(dmList.get()));
        					backUpByConditions(taskStagePO.get(),srcTablePO.get());
        				}catch(Exception e){
        					LOGGER.error("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"|备份线程异常|",e);
        					errorMsg.set("备份线程异常"+e.toString());
        					Flag.set(false);
        				}
        				//判断该批次是否执行成功
        				if(!Flag.get()){
                    		LOGGER.error("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"执行错误更新体检结果明细");
        					taskStageState.set(DMSConstant.Status.FALSE.toString());
        					//更新体检结果明细
        					DmPO record = new DmPO();
        					record.setState(DMSConstant.Status.BACKUPFALSE.toString());
        					record.setErrorDesc(errorMsg.get());
        					record.setStateDate(DateUtils.getNowDate());
        					updateByIds(record, "DM_"+taskStagePO.get().getCheckObjName(),keyList.get());
        					//更新任务环节表（dm_task_stage）的状态，状态时间，任务结束时间
                    		LOGGER.error("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"执行错误更新任务环节表");
        					taskStagePO.set(dmTaskStageDao.selectByPrimaryKey(taskStagePO.get().getDmInfoId()));
        					taskStagePO.get().setState(DMSConstant.Status.FALSE.toString());
        					taskStagePO.get().setErrorDesc(errorMsg.get());
        					taskStagePO.get().setStateDate(DateUtils.getNowDate());
        					taskStagePO.get().setTaskEndTime(DateUtils.getNowDate());
        					dmTaskStageDao.updateByPrimaryKey(taskStagePO.get());
        					//更新清理的任务环节表
        					DmTaskStageArg stageArg = new DmTaskStageArg();
        					DmTaskStageArg.DmTaskStageCriteria stageC = stageArg.createCriteria();
        					stageC.andDmTaskIdEqualTo(taskStagePO.get().getDmTaskId());
        					stageC.andDmStrategyIdEqualTo(taskStagePO.get().getDmStrategyId());
        					stageC.andStageEqualTo(DMSConstant.STAGE_CLEANUP);
        					stageC.andCheckObjNameEqualTo(taskStagePO.get().getCheckObjName());//
        					DmTaskStagePO stagePo = new DmTaskStagePO();
        					stagePo.setState(DMSConstant.Status.FALSE.toString());
        					stagePo.setErrorDesc("备份失败");
        					stagePo.setStateDate(DateUtils.getNowDate());
//        					//调用清理  modified by wb.l 备份失败发告警信息~不做任何程序清理动作
//                    		LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"执行错误调用清理");
//        					cleanUpState = cleanUpService.cleanUp(taskStagePO.get().getDmTaskId(), taskStagePO.get().getDmStrategyId(), DMSConstant.STAGE_BACKUP);
//    						if (cleanUpState != null) {		
//                        		LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"执行错误更新清理的任务环节表");
//    							dmTaskStageDao.updateByArgSelective(stagePo, stageArg);
//    						}
        					dmTaskStageDao.updateByArgSelective(stagePo, stageArg);
        					LOGGER.error("备份失败托送告警信息-----");
        					resourceCheckJob.pushMsg2IOCP("[backup error]strategyID:"+taskStagePO.get().getDmStrategyId(), errorMsg.get(), "备份策略"+taskStagePO.get().getDmStrategyId()+"失败");
        					
        					
        					//处理从表任务环节表的状态，状态时间，任务结束时间
        					if (DMSConstant.ARCHIVE_TYPE_REL==srcTablePO.get().getArchiveType()) {
                        		LOGGER.error("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"执行错误更新从表任务环节表");
        						DmSecondaryStageArg secondaryStageArg = new DmSecondaryStageArg();
        						DmSecondaryStageArg.DmSecondaryStageCriteria secondaryStageCriteria = secondaryStageArg.createCriteria();
        						secondaryStageCriteria.andDmInfoIdEqualTo(taskStagePO.get().getDmInfoId());
        						secondaryStageCriteria.andStateEqualTo(DMSConstant.Status.PROCESSING.toString());
                				secondaryStageCriteria.andStageEqualTo(DMSConstant.STAGE_BACKUP);
        						secondaryStageCriteria.andDmTaskIdEqualTo(taskStagePO.get().getDmTaskId());
        						secondaryStageCriteria.andDmStrategyIdEqualTo(taskStagePO.get().getDmStrategyId());
        						DmSecondaryStagePO secondaryStagePO = new DmSecondaryStagePO();
        						secondaryStagePO.setState(DMSConstant.Status.FALSE.toString());
        						secondaryStagePO.setStateDate(DateUtils.getNowDate());
        						secondaryStagePO.setTaskEndTime(DateUtils.getNowDate());
        						secondaryStageDao.updateByArgSelective(secondaryStagePO, secondaryStageArg);
        					}
        					continueFlag.set(false);
        					return "00F";
        				}else{
        					//更新体检结果明细
        					//(先全部更新为失败，再将查到的数据更新为成功)
                    		LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"执行成功更新体检结果明细");
        					int counts = 0;
        					DmPO record = new DmPO();
        					record.setState(DMSConstant.Status.BACKUPFALSE.toString());
        					record.setErrorDesc("在源数据库找不到数据");
        					record.setStateDate(DateUtils.getNowDate());
        					updateByIds(record, "DM_"+taskStagePO.get().getCheckObjName(),keyList.get());
        					String tableName = null;
        					List<Long> backUpList = null;
        					for (Entry<String, List<Long>> dmEntryList : backUpMap.get().entrySet()) {
        						tableName = dmEntryList.getKey();
        						backUpList = dmEntryList.getValue();
        						record.setState(DMSConstant.Status.TOCLEANUP.toString());
        						record.setErrorDesc("");
        						record.setDmTargetTableName(tableName);
        						updateByIds(record, "DM_"+taskStagePO.get().getCheckObjName(),backUpList);
        						counts += backUpList.size();
        					}
        					//更新任务环节表的备份数据量，备份数据大小
                    		LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"执行成功更新任务环节表的备份数据量，备份数据大小");
        					taskStagePO.set(dmTaskStageDao.selectByPrimaryKey(taskStagePO.get().getDmInfoId()));
        					int count = 0;
        					if(taskStagePO.get().getCheckCount()==null){
        						count = counts;
        					}else{
        						count = taskStagePO.get().getCheckCount()+counts;
        					}
        					BigDecimal d = taskStagePO.get().getCheckDataSize();
        					if(d==null){
        						d = BigDecimal.ZERO;
        					}
        					BigDecimal size =d.add(BigDecimal.valueOf(Double.valueOf(counts*targetTablePO.get().getOneRecordSize())/1024).setScale(2, RoundingMode.HALF_UP));
        					taskStagePO.get().setCheckCount(count);
        					taskStagePO.get().setCheckDataSize(size);
        					dmTaskStageDao.updateByPrimaryKey(taskStagePO.get());
        				}
        				//更新从表任务环节表备份数据量，备份数据大小
        				if (DMSConstant.ARCHIVE_TYPE_REL==srcTablePO.get().getArchiveType()) {
                    		LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"执行成功更新从表任务环节表备份数据量，备份数据大小");
        					Map<String,Integer>tempMap = new HashMap<String, Integer>();
        					List< DmSecondaryStagePO> tempList;
        					Integer sourceDataId = null;
        					String tableName = null;
        					int count;
        					int size;
        					for (Entry<Integer, Map<String, Integer>> dmEntryList : childSourceNameMap.get().entrySet()) {
        						sourceDataId = dmEntryList.getKey();
        						tempMap = dmEntryList.getValue();
        						size = childRecordSizeMap.get().get(sourceDataId);
        						for (Entry<String, Integer> tableEntry : tempMap.entrySet()) {
        							tableName = tableEntry.getKey();
        							count = tableEntry.getValue();
        							//更新备份&清理
        							DmSecondaryStageArg secondaryStageArg = new DmSecondaryStageArg();
        							DmSecondaryStageArg.DmSecondaryStageCriteria secondaryStageCriteria = secondaryStageArg.createCriteria();
        							List<Integer> infoIds = new ArrayList<Integer>();
        							infoIds.add(clearPO.get().getDmInfoId());
        							infoIds.add(taskStagePO.get().getDmInfoId());
        							secondaryStageCriteria.andDmInfoIdIn(infoIds);
        							secondaryStageCriteria.andSourceDataIdEqualTo(sourceDataId);
        							secondaryStageCriteria.andDmTaskIdEqualTo(taskStagePO.get().getDmTaskId());
        							secondaryStageCriteria.andDmStrategyIdEqualTo(taskStagePO.get().getDmStrategyId());
        							secondaryStageCriteria.andCheckObjNameEqualTo(tableName);
        							tempList = secondaryStageDao.selectByArg(secondaryStageArg);
        							for (DmSecondaryStagePO record : tempList) {
        								count += record.getResultCount();
        								record.setTotalCount(count);
        								record.setTotalSize(BigDecimal.valueOf(Double.valueOf(count*size)/1024).setScale(2,RoundingMode.HALF_UP));
        								if(DMSConstant.STAGE_BACKUP==record.getStage()){
        									record.setState(DMSConstant.Status.SUCCESS.toString());
        									record.setTaskEndTime(DateUtils.getNowDate());
        									record.setResultCount(count);
        									record.setResultSize(BigDecimal.valueOf(Double.valueOf(count*size)/1024).setScale(2,RoundingMode.HALF_UP));
        								}
        								secondaryStageDao.updateByPrimaryKey(record);
        							}
        						}
        						
        					}
        				}
        			} else if (archiveMode == DMSConstant.ARCHIVE_MODE_SUBAREA) {
        				// 2：按分区归档；
        			} else if (archiveMode == DMSConstant.ARCHIVE_MODE_WHOLETABLE) {
        				// 3：按整表归档；
        			}
        		}
        	}
        }
       if("".equals( taskStageState.get())){
    	   return DMSConstant.Status.SUCCESS.toString();
       }else{
    	   return taskStageState.get();
       }
    }
    
    /**
     * 根据明细主键分批次更新体检明细
     * @param record
     * @param tableName
     * @param keyList2
     * @throws BaseAppException
     */
    private void updateByIds(DmPO record, String tableName, List<Long> keyList) throws BaseAppException {
    	 List<List<Long>> listArray = new ArrayList<List<Long>>();
         int pageSize = 500 ;
         ArrayList<Long> al = new ArrayList<Long>();
         for(Long x : keyList){
             al.add(x);
             if (pageSize == al.size()){
                 listArray.add(al);
                 al = new ArrayList<Long>();
             }
         }
         if (0 != al.size()){
             listArray.add(al);
         }
		for (List<Long> subList : listArray) {
			DmArg dmArg = new DmArg();
			DmArg.DmCriteria c = dmArg.createCriteria();
			c.andDmDetailIdIn(subList);
			dmDao.updateByArg(record, tableName, dmArg);
		}
		
	}



	/**
     * 条件归档
     * @param taskStagePO 任务环节实体
     * @param dmList 体检结果明细列表
     * @throws Exception
     */
    private  void backUpByConditions(DmTaskStagePO taskStagePO,DmSrcTablePO srcTablePO){
		//拼接源数据查询语句
		LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"拼接源数据查询语句");
		StringBuffer SQL;
		//源表名
		String tableName = taskStagePO.getCheckObjName();
		//查询字段
		//Select distinct source_column from target_col_map where source_data_id=? And state=’00A’ And source_col_flag=’T
		LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"拼接源数据查询字段");
		String searchCols = getSearchColsBySourceId(srcTablePO.getSourceDataId(),sourceDataTypeMap.get());
		if("false".equals(searchCols)){
			Flag.set(false);
			LOGGER.error("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"|目标映射表未配置|");
			errorMsg.set("目标映射表未配置");
		}
		Map<String,Object> tempMap = new HashMap<String, Object>();
		PreparedStatement stmt = null;
		ResultSet rs = null;
		DataSource sourceDs = null;
		Connection con = null;
		DmPO dmPO = null;
		//源数据库信息
		//连接源数据库
		try {
			LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"连接源数据库");
			String[] dbConfig =  baseInfoService.getDbConfig(srcTablePO.getDbLinkId(), srcTablePO.getDbUserName());
			sourceDs = dynamicDataSourceManager.getDruidDataSource(srcTablePO.getDbLinkId()+srcTablePO.getDbUserName(), dbConfig);
			LOGGER.debug(Thread.currentThread()+"获取链接信息中...");
			con = DataSourceUtils.getConnection(sourceDs);
			LOGGER.debug(Thread.currentThread()+"获取链接信息ok...");
			if(DMSConstant.DB_LINK_CATEGORY_INFORMIX.equals(dbConfig[0].toUpperCase())){
				con.setTransactionIsolation(Connection.TRANSACTION_READ_UNCOMMITTED);
			}
			//拼接源表数据查询查询语句
			LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"查询源数据");
			for (int k = 0; k < dmList.get().size(); k++) {
				dmPO = dmList.get().get(k);
				SQL = new StringBuffer();
				SQL.append("select " + dmPO.getDmDetailId() + "  as detail_id");
				SQL.append(searchCols).append(" from ").append(tableName + " ").append(WhereCriteria(dmPO.getDmObjectKeyName()));
				String[] colName = searchCols.split(",");
				//查询源数据
				if (k == 0) {//输出查询语句					
					LOGGER.info("{1}查询源数据{0}",SQL.toString(), Thread.currentThread());
				}
				stmt = con.prepareStatement(SQL.toString());
				this.pstSetObject(stmt,dmPO.getDmObjectKeyValue(),dmPO.getDmObjectKeyName(),sourceDataTypeMap.get(),null,dbConfig[0]);
				rs = stmt.executeQuery();
				tempMap = new HashMap<String, Object>();
				while (rs.next()) {
					for (int i=1;i<colName.length;i++) {
						tempMap.put(colName[i], rs.getObject(i+1));
					}
				}
				sourceResultMap.get().put(dmPO.getDmDetailId(), tempMap);
				JdbcUtils.closeStatement(stmt);
				JdbcUtils.closeResultSet(rs);
			}
		} catch (Exception e) {
			Flag.set(false);
			LOGGER.error("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"|获取源数据连接信息异常|",e);
			errorMsg.set("执行源表数据查询异常"+e.toString());
			return  ;
		}finally{
			JdbcUtils.closeStatement(stmt);
			JdbcUtils.closeResultSet(rs);
			DataSourceUtils.releaseConnection(con, sourceDs);
		}
		//查找插入目标表字段与源表字段的对应关系
		LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"查找插入目标表字段与源表字段的对应关系");
		TargetColMapArg arg = new TargetColMapArg();
		TargetColMapArg.TargetColMapCriteria cri = arg.createCriteria();
		cri.andSourceDataIdEqualTo(srcTablePO.getSourceDataId());
		cri.andTargetDataIdEqualTo(targetTablePO.get().getTargetDataId());
		List<TargetColMapPO> colMapList = targetColMapDao.selectByArg(arg);
		String targetTableName;
		List<Long> tempList ;
		Map<String, Object> sourceDataMap;
		LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"拼接目标表名");
		for (int k = 0; k < dmList.get().size(); k++) {
			dmPO = dmList.get().get(k);
			sourceDataMap = sourceResultMap.get().get(dmPO.getDmDetailId());
			if(!Utils.isEmpty(sourceDataMap)){
				//拼接目标表名
				targetTableName = getTableName(targetTablePO.get().getTableName(),targetTablePO.get().getSubFlag(),
						targetTablePO.get().getTargetDataId(),2, srcTablePO,sourceDataMap);
				dmPO.setDmTargetTableName(targetTableName);//设置备份目标表名
				if(!Utils.isEmpty(backUpMap)&&backUpMap.get().keySet().contains(targetTableName)){
					backUpMap.get().get(targetTableName).add(dmPO.getDmDetailId());
				}else{
					tempList =new ArrayList<Long>();
					tempList.add(dmPO.getDmDetailId());
					backUpMap.get().put(targetTableName,tempList);
				}
			}
		}
		DataSource targetDs = null;
		Connection conn = null;
		PreparedStatement pstmt = null;
		try {
			LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"获取目标数据源");
			String[] targetDbConfig =  baseInfoService.getDbConfig(targetTablePO.get().getDbLinkId(), targetTablePO.get().getDbUserName());
			targetDs = dynamicDataSourceManager.getDruidDataSource(targetTablePO.get().getDbLinkId()+targetTablePO.get().getDbUserName(), targetDbConfig);
			LOGGER.debug(Thread.currentThread()+"||获取数据库连接中...");
			conn = DataSourceUtils.getConnection(targetDs);
			LOGGER.debug(Thread.currentThread()+"获取链接信息ok...");
			conn.setAutoCommit(false);
			//拼接目标表插入语句,执行插入语句
			LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"拼接目标表插入语句,执行插入语句");
			for (String targetName : backUpMap.get().keySet()) {
				//分表批量插入数据
				LOGGER.info("{1}批量插入目标表{0}",targetName,Thread.currentThread());
				pstmt=InsertBatch(conn,targetName,backUpMap.get(),colMapList,sourceResultMap.get());
			    pstmt.executeBatch();
			}
			conn.commit();
		} catch (Exception e) {
			Flag.set(false);
			LOGGER.error("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"|备份主表信息异常|",e);
			errorMsg.set("备份主表信息异常："+e.toString());
			try {
				conn.rollback();
			} catch (SQLException e1) {
			}
		} finally {
			JdbcUtils.closeStatement(pstmt);
			DataSourceUtils.releaseConnection(conn, targetDs);
		}
		//判断是否有分表要归档
		if (DMSConstant.ARCHIVE_TYPE_REL==srcTablePO.getArchiveType()) {
			//关联归档
			//查询关联归档数据
			LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"查询关联归档数据");
			childSourceList.set(srcTableCustomDao.selectByStrategyRef(taskStagePO.getDmStrategyId()));
			List<DmSecondaryStagePO> saveList = new ArrayList<DmSecondaryStagePO>();
			DmSecondaryStagePO record;
			childRecordSizeMap.set(new HashMap<Integer, Integer>());
			String childSearchCols  ;
			StringBuffer childSQL;
			String targetChildTableName;
			List<Map<String, Object>> subObjetList ;
			//Map<String, Object> childSourceDataMap;
			List<Long> backUpList ;
			Map <String,Integer> tempNameMap = new HashMap<String, Integer>();
			DataSource sourceChildDs = null;
			Connection conChild = null;
			DataSource targetChildDs = null;
			Connection connTargetChild = null;
			List<Map<String, Object>> tempO = null;
			Map<Integer,Map<Long,String>> dmTablename = new HashMap<Integer, Map<Long,String>>();
			Map<Integer,Map<String,List<Long>>> subtarname = new HashMap<Integer, Map<String,List<Long>>>();
			for (DmSrcTablePO srcPO : childSourceList.get()) { 
				Map<Long,String> subTablename = new HashMap<Long, String>();
				//连接源数据库
				LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"获取源从表数据源");
				try {
					String[] dbChildConfig = baseInfoService.getDbConfig(srcPO.getDbLinkId(), srcPO.getDbUserName());
					sourceChildDs = dynamicDataSourceManager.getDruidDataSource(srcPO.getDbLinkId()+ srcPO.getDbUserName(), dbChildConfig);
					LOGGER.debug(Thread.currentThread()+" ||获取数据库连接中...");
					conChild = DataSourceUtils.getConnection(sourceChildDs);
					LOGGER.debug(Thread.currentThread()+"获取链接信息ok...");
					if(DMSConstant.DB_LINK_CATEGORY_INFORMIX.equals(dbChildConfig[0].toUpperCase())){
						conChild.setTransactionIsolation(Connection.TRANSACTION_READ_UNCOMMITTED);
					}
					childRecordSizeMap.get().put(srcPO.getSourceDataId(), srcPO.getOneRecordSize());
					backUpList = new ArrayList<Long>();
					//拼接从表查询字段
					LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"拼接从表查询字段");
					childSearchCols = getSearchColsBySourceId(srcPO.getSourceDataId(),childSourceDataTypeMap.get());
					//查找where条件的关键字段
					LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"查找where条件的关键字段");
					DmTableRelArg tableRelArg = new DmTableRelArg();
					DmTableRelArg.DmTableRelCriteria tableRelCriteria = tableRelArg.createCriteria();
					tableRelCriteria.andStateEqualTo(DMSConstant.Status.SUCCESS.toString());
					tableRelCriteria.andRelDataIdEqualTo(srcTablePO.getSourceDataId());
					tableRelCriteria.andSubDataIdEqualTo(srcPO.getSourceDataId());
					DmTableRelPO tableRelPO = dmTableRelDao.selectByArg(tableRelArg).get(0);
					String childSourceName = "";
					//将查询结果放入childResultMap中。Interger为detailId，String为字段名，Object为字段值
					childSourceResultMap.set(new HashMap<Long, List<Map<String,Object>>>());
					LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"拼接源从表表名");
					for (int k = 0; k < dmList.get().size(); k++) {
						dmPO = dmList.get().get(k);
						sourceDataMap = sourceResultMap.get().get(dmPO.getDmDetailId());
						tempNameMap = childSourceNameMap.get().get(srcPO.getSourceDataId());
						childSourceName = getTableName(srcPO.getTableName(),srcPO.getSubFlag(),srcPO.getSourceDataId(),1,
								srcTablePO,sourceDataMap);
						subTablename.put(dmPO.getDmDetailId(), childSourceName);
						if(Utils.isEmpty(tempNameMap)||!tempNameMap.keySet().contains(childSourceName)){
							tempNameMap = new HashMap<String, Integer>();
							tempNameMap.put(childSourceName, 1);
							childSourceNameMap.get().put(srcPO.getSourceDataId(), tempNameMap);
							DmSecondaryStageArg secondaryStageArg = new DmSecondaryStageArg();
							DmSecondaryStageArg.DmSecondaryStageCriteria secondaryStageCriteria = secondaryStageArg.createCriteria();
							secondaryStageCriteria.andDmInfoIdEqualTo(taskStagePO.getDmInfoId());
							secondaryStageCriteria.andSourceDataIdEqualTo(srcPO.getSourceDataId());
							secondaryStageCriteria.andDmTaskIdEqualTo(taskStagePO.getDmTaskId());
							secondaryStageCriteria.andDmStrategyIdEqualTo(taskStagePO.getDmStrategyId());
							secondaryStageCriteria.andCheckObjNameEqualTo(childSourceName);
							List<DmSecondaryStagePO>tList = secondaryStageDao.selectByArg(secondaryStageArg);
							if(Utils.isEmpty(tList)){
								LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"创建从表环节记录");
								//创建备份&清理的记录，放入saveList
								record = new DmSecondaryStagePO();
								record.setDmSecondaryInfoId(sequenceGenerator.sequenceIntValue("DM_SECONDARY_STAGE","DM_SECONDARY_INFO_ID"));
								record.setDmTaskId(taskStagePO.getDmTaskId());
								record.setDmStrategyId(taskStagePO.getDmStrategyId());
								record.setDmInfoId(taskStagePO.getDmInfoId());
								record.setSourceDataId(srcPO.getSourceDataId());
								record.setCheckObjName(childSourceName);
								record.setStage(DMSConstant.STAGE_BACKUP);
								record.setState(DMSConstant.Status.PROCESSING.toString());
								record.setStateDate(DateUtils.getNowDate());
								record.setTaskStartTime(DateUtils.getNowDate());
								record.setCreateDate(DateUtils.getNowDate());
								record.setNodeId(taskStagePO.getNodeId());
								saveList.add(record);
								record = new DmSecondaryStagePO();
								record.setDmSecondaryInfoId(sequenceGenerator.sequenceIntValue("DM_SECONDARY_STAGE","DM_SECONDARY_INFO_ID"));
								record.setStage(DMSConstant.STAGE_CLEANUP);
								record.setState(DMSConstant.Status.PENDING.toString());
								record.setDmTaskId(taskStagePO.getDmTaskId());
								record.setDmStrategyId(taskStagePO.getDmStrategyId());
								record.setDmInfoId(clearPO.get().getDmInfoId());
								record.setSourceDataId(srcPO.getSourceDataId());
								record.setCheckObjName(childSourceName);
								record.setStateDate(DateUtils.getNowDate());
								record.setTaskStartTime(null);
								record.setCreateDate(DateUtils.getNowDate());
								record.setNodeId(taskStagePO.getNodeId());
								saveList.add(record);
							}
						}else{
							int t = tempNameMap.get(childSourceName);
							tempNameMap.put(childSourceName, t+1);
							childSourceNameMap.get().put(srcPO.getSourceDataId(), tempNameMap);
						}
						//LOGGER.debug("拼接源从表查询语句"+ dmPO.getDmDetailId());
						childSQL = new StringBuffer();
						childSQL.append("select " + dmPO.getDmDetailId() + "  as detail_id");
						childSQL.append(childSearchCols);
						childSQL.append(" from ");
						childSQL.append(childSourceName );
						childSQL.append(WhereCriteria(tableRelPO.getSubMapCols()));
						//LOGGER.debug("获取从表源数据信息");
						stmt = conChild.prepareStatement(childSQL.toString());
						this.pstSetObject(stmt,tableRelPO.getRelMapCols(),tableRelPO.getSubMapCols(),sourceDataTypeMap.get(),sourceDataMap,dbChildConfig[0]);
						String[] colName = childSearchCols.split(",");
						rs = stmt.executeQuery();
						tempMap = new HashMap<String, Object>();
						tempO = new ArrayList<Map<String,Object>>();
						while (rs.next()) {
							for (int i=1;i<colName.length;i++) {
								tempMap.put(colName[i], rs.getObject(i+1));
							}
							tempO.add(tempMap);
						}
						if(tempO.size()==0){
							int t = tempNameMap.get(childSourceName);
							tempNameMap.put(childSourceName, t-1);
							childSourceNameMap.get().put(srcPO.getSourceDataId(), tempNameMap);
						}
						childSourceResultMap.get().put(dmPO.getDmDetailId(), tempO);
						JdbcUtils.closeStatement(stmt);
						JdbcUtils.closeResultSet(rs);
					}
					dmTablename.put(srcPO.getSourceDataId(), subTablename);
				} catch (Exception e) {
					Flag.set(false);
					try {
						conChild.rollback();
					} catch (SQLException e1) {
						Flag.set(false);
						LOGGER.error("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"|备份主表释放数据连接异常|",e1);
						errorMsg.set("备份主表释放数据连接异常："+e1.toString());
					}
					LOGGER.error("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"|获取从表源数据信息异常|",e);
					errorMsg.set("获取从表源数据信息异常"+e.toString());
					return  ;
				}finally{
					JdbcUtils.closeStatement(stmt);
					JdbcUtils.closeResultSet(rs);
					DataSourceUtils.releaseConnection(conChild, sourceChildDs);
				}
				//查询从表目标对象
				LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"查询从表目标对象");
				DmTargetTableArg targetTableArg = new DmTargetTableArg();
				DmTargetTableArg.DmTargetTableCriteria tableCriteria = targetTableArg.createCriteria();
				tableCriteria.andSourceDataIdEqualTo(srcPO.getSourceDataId());
				DmTargetTablePO childTargetTablePO = targetTableDao.selectByArg(targetTableArg).get(0);
				//连接源数据库
				try {
					LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"获取从表目标数据源");
					String[] dbTargetChildConfig = baseInfoService.getDbConfig(childTargetTablePO.getDbLinkId(), childTargetTablePO.getDbUserName());
					targetChildDs = dynamicDataSourceManager.getDruidDataSource(childTargetTablePO.getDbLinkId()+childTargetTablePO.getDbUserName(), dbTargetChildConfig);
					LOGGER.debug(Thread.currentThread()+" ||获取数据库连接中...");
					connTargetChild = DataSourceUtils.getConnection(targetChildDs);
					LOGGER.debug(Thread.currentThread()+"获取链接信息ok...");
					connTargetChild.setAutoCommit(false);
					//查找插入目标表字段与源表字段的对应关系
					LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"查找插入目标表字段与源表字段的对应关系");
					TargetColMapArg colArg = new TargetColMapArg();
					TargetColMapArg.TargetColMapCriteria colCriteria = colArg.createCriteria();
					colCriteria.andSourceDataIdEqualTo(srcPO.getSourceDataId());
					colCriteria.andTargetDataIdEqualTo(childTargetTablePO.getTargetDataId());
					List<TargetColMapPO> childColMapList = targetColMapDao.selectByArg(colArg);
					backUpMap.set(new HashMap<String, List<Long>>());
					//拼接目标表插入语句
					LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"拼接目标表插入语句");
					LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"拼接目标表名");
					for (DmPO dmPo : dmList.get()) {
						subObjetList = childSourceResultMap.get().get(dmPo.getDmDetailId());
						for (Map<String, Object> map : subObjetList) {
							if(!Utils.isEmpty(map)){
								backUpList.add(dmPo.getDmDetailId());
								//拼接目标表名
								targetChildTableName = getTableName(childTargetTablePO.getTableName(),childTargetTablePO.getSubFlag(),
										childTargetTablePO.getTargetDataId(), 2,srcPO,map);
								if(!Utils.isEmpty(backUpMap)&&backUpMap.get().keySet().contains(targetChildTableName)){
									if(!backUpMap.get().get(targetChildTableName).contains(dmPo.getDmDetailId())){
										backUpMap.get().get(targetChildTableName).add(dmPo.getDmDetailId());
									}
								}else{
									tempList =new ArrayList<Long>();
									tempList.add(dmPo.getDmDetailId());
									backUpMap.get().put(targetChildTableName,tempList);
								}
							}
						}
					}
					subtarname.put(srcPO.getSourceDataId(), backUpMap.get());
					//拼接目标表插入语句,执行插入语句
					LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"拼接目标表插入语句,执行插入语句");
					for (String targetName : backUpMap.get().keySet()) {
						//分表批量插入数据
						pstmt=InsertBatchForSub(connTargetChild,targetName,backUpMap.get(),childColMapList,childSourceResultMap.get());
					    pstmt.executeBatch();
					}
					connTargetChild.commit();
				} catch (Exception e) {
					Flag.set(false);
					try {
						connTargetChild.rollback();
					} catch (SQLException e1) {
						Flag.set(false);
						LOGGER.error("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"|备份主表释放数据连接异常|",e1);
						errorMsg.set("备份主表释放数据连接异常："+e1.toString());
					}
					LOGGER.error("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"备份从表信息异常|",e);
					errorMsg.set("备份从表信息异常"+e.toString());
					return;
				}finally{
					//释放从表连接资源
					JdbcUtils.closeStatement(pstmt);
					DataSourceUtils.releaseConnection(connTargetChild, targetChildDs);
				}
				//生成从表的任务环节
				if(!Utils.isEmpty(saveList)){
					LOGGER.debug("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"生成从表的任务环节");
					secondaryStageDao.insertBatch(saveList);
					saveList = new ArrayList<DmSecondaryStagePO>();
				}
				allBackUpTable.get().add(srcPO);
			}
		} 
//		//commit数据，关闭连接
//		if(Flag.get()){
//			try {
//				conn.rollback();
//			} catch (SQLException e) {
//				Flag.set(false);
//				LOGGER.error("任务Id："+taskStagePO.getDmTaskId()+",策略Id:"+taskStagePO.getDmStrategyId()+"|备份主表回滚异常|",e);
//				errorMsg.set("备份主表回滚异常："+e.toString());
//			} finally {
//				JdbcUtils.closeStatement(pstmt);
//				DataSourceUtils.releaseConnection(conn, targetDs);
//			}
//		}
    }
    
    /**
     * 删除已备份的分表
     * @param list
     * @param dmTablename 
     * @param subtarname 
     */
    private void removeData(List<DmSrcTablePO> list, Map<Integer, Map<Long, String>> dmTablename, Map<Integer, Map<String, List<Long>>> subtarname) {
    	DataSource sourceChildDs = null;
		Connection conChild = null;
		DataSource targetChildDs = null;
		Connection connTargetChild = null; 
		PreparedStatement stmt = null;
		ResultSet rs = null;
		String childSearchCols = "";
		StringBuffer childSQL;
		Map<String,Object> tempMap;
		String[] dbChildConfig;
		List<Map<String,Object>> tempO;
		Map<Long,String> childSourceName;
		Map<String,Object>sourceDataMap;
		Map<String,Object> childSourceDataMap;
		for (DmSrcTablePO srcPO : list) {
			//连接源数据库
			LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"获取源从表数据源");
			try {
				dbChildConfig =  baseInfoService.getDbConfig(srcPO.getDbLinkId(), srcPO.getDbUserName());
				sourceChildDs = dynamicDataSourceManager.getDruidDataSource(srcPO.getDbLinkId()+srcPO.getDbUserName(), dbChildConfig);
				LOGGER.debug(Thread.currentThread()+"||获取数据库连接中...");
				conChild = DataSourceUtils.getConnection(sourceChildDs);
				LOGGER.debug(Thread.currentThread()+"获取链接信息ok...");
				if(DMSConstant.DB_LINK_CATEGORY_INFORMIX.equals(dbChildConfig[0].toUpperCase())){
					conChild.setTransactionIsolation(Connection.TRANSACTION_READ_UNCOMMITTED);
				}
				//拼接从表查询字段
				LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"拼接从表查询字段");
				childSearchCols = getSearchColsBySourceId(srcPO.getSourceDataId(),childSourceDataTypeMap.get());
				//查找where条件的关键字段
				LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"查找where条件的关键字段");
				DmTableRelArg tableRelArg = new DmTableRelArg();
				DmTableRelArg.DmTableRelCriteria tableRelCriteria = tableRelArg.createCriteria();
				tableRelCriteria.andStateEqualTo(DMSConstant.Status.SUCCESS.toString());
				tableRelCriteria.andRelDataIdEqualTo(srcTablePO.get().getSourceDataId());
				tableRelCriteria.andSubDataIdEqualTo(srcPO.getSourceDataId());
				DmTableRelPO tableRelPO = dmTableRelDao.selectByArg(tableRelArg).get(0);
				childSourceName = dmTablename.get(srcPO.getSourceDataId());
				//将查询结果放入childResultMap中。Interger为detailId，String为字段名，Object为字段值
				childSourceResultMap.set(new HashMap<Long,List<Map<String,Object>>>());
				LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"拼接源从表表名");
				for(DmPO dmPO: dmList.get()){
					sourceDataMap = sourceResultMap.get().get(dmPO.getDmDetailId());
					//LOGGER.debug("拼接源从表查询语句");
					childSQL = new StringBuffer();
					childSQL.append("select " + dmPO.getDmDetailId() + "  as detail_id");
					childSQL.append(childSearchCols);
					childSQL.append(" from ");
					childSQL.append(childSourceName.get(dmPO.getDmDetailId()));
					childSQL.append(WhereCriteria(tableRelPO.getSubMapCols()));
					//LOGGER.debug("获取从表源数据信息");
					stmt = conChild.prepareStatement(childSQL.toString());
					this.pstSetObject(stmt,tableRelPO.getRelMapCols(),tableRelPO.getSubMapCols(),sourceDataTypeMap.get(),sourceDataMap,dbChildConfig[0]);
					String[] colName = childSearchCols.split(",");
					rs = stmt.executeQuery();
					tempMap = new HashMap<String, Object>();
					tempO = new ArrayList<Map<String,Object>>();
					while (rs.next()) {
						for (int i=1;i<colName.length;i++) {
							tempMap.put(colName[i], rs.getObject(i+1));
						}
						tempO.add(tempMap);
					}
					childSourceResultMap.get().put(dmPO.getDmDetailId(), tempO);
				}
			} catch (Exception e) {
				Flag.set(false);
				LOGGER.error("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"|获取从表源数据信息异常|",e);
				errorMsg.set("获取从表源数据信息异常"+e.toString());
				return  ;
			}finally{
				JdbcUtils.closeStatement(stmt);
				JdbcUtils.closeResultSet(rs);
				try {
					DataSourceUtils.doReleaseConnection(conChild, sourceChildDs);
				} catch (SQLException e) {
					Flag.set(false);
					LOGGER.error("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"|释放从表源数据连接异常|",e);
					errorMsg.set("释放从表源数据连接异常"+e.toString());
					return  ;
				}
			}
			//查询从表目标对象
			LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"查询从表目标对象");
			DmTargetTableArg targetTableArg = new DmTargetTableArg();
			DmTargetTableArg.DmTargetTableCriteria tableCriteria = targetTableArg.createCriteria();
			tableCriteria.andSourceDataIdEqualTo(srcPO.getSourceDataId());
			DmTargetTablePO childTargetTablePO = targetTableDao.selectByArg(targetTableArg).get(0);
			String[] dbTargetChildConfig;
			//连接源数据库
			try {
				LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"获取从表目标数据源");
				dbTargetChildConfig =  baseInfoService.getDbConfig(childTargetTablePO.getDbLinkId(), childTargetTablePO.getDbUserName());
				targetChildDs = dynamicDataSourceManager.getDruidDataSource(childTargetTablePO.getDbLinkId()+childTargetTablePO.getDbUserName(), dbTargetChildConfig);
				LOGGER.debug(Thread.currentThread()+"||获取数据库连接中...");
				connTargetChild = DataSourceUtils.getConnection(targetChildDs);
				LOGGER.debug(Thread.currentThread()+"获取链接信息ok...");
				connTargetChild.setAutoCommit(false);
				//查找插入目标表字段与源表字段的对应关系
				LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"查找插入目标表字段与源表字段的对应关系");
				TargetColMapArg colArg = new TargetColMapArg();
				TargetColMapArg.TargetColMapCriteria colCriteria = colArg.createCriteria();
				colCriteria.andSourceDataIdEqualTo(srcPO.getSourceDataId());
				colCriteria.andTargetDataIdEqualTo(childTargetTablePO.getTargetDataId());
				List<TargetColMapPO> childColMapList = targetColMapDao.selectByArg(colArg);
				//拼接目标表删除语句
				LOGGER.debug("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"拼接目标表删除语句");
				for (String targetName : backUpMap.get().keySet()) {
					//分表批量插入数据
					stmt=deleteBatchForSub(connTargetChild,targetName,subtarname.get(srcPO.getSourceDataId()),childColMapList,childSourceResultMap.get());
				    stmt.executeBatch();
				}
				try {
					connTargetChild.commit();
				} catch (Exception e) {
					connTargetChild.rollback();
				}
			} catch (Exception e) {
				Flag.set(false);
				LOGGER.error("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"|删除从表信息异常|",e);
				errorMsg.set("删除从表信息异常"+e.toString());
				return;
			}finally{
				JdbcUtils.closeStatement(stmt);
				try {
					DataSourceUtils.doReleaseConnection(connTargetChild, targetChildDs);
				} catch (SQLException e) {
					Flag.set(false);
					LOGGER.error("任务Id："+taskStagePO.get().getDmTaskId()+",策略Id:"+taskStagePO.get().getDmStrategyId()+"|释放从表目标数据连接异常|",e);
					errorMsg.set("释放从表目标数据连接异常"+e.toString());
					return  ;
				}
			}
		}
	}

	/**
     * 拼接where查询条件
     * @param name 关键字段
     * @return where语句
     */
    private String WhereCriteria(String name){
//    	StringBuffer sb = new StringBuffer(" where 1=1 ");
//    	String[] nameStrs = name.split(",");
//    	for(int i=0;i<nameStrs.length;i++){
//    		sb.append(" and "+nameStrs[i]+" = ? ");
//    	}
    	StringBuffer sb = new StringBuffer(" where ");
    	String[] nameStrs = name.split(",");
    	for(int i=0, j = nameStrs.length; i < j; i++){
    		if (i!=0) {
    			sb.append(" and ");
    		}
    		sb.append(nameStrs[i]+" = ? ");
    	}
    	return sb.toString();
    }
    
    /**
     * 拼接目标表插入语句
     * @param targetTableName 目标表表名
     * @param colMapList 源字段和目标字段的映射
     * @return
     */
    private String insertSql(String targetTableName,List<TargetColMapPO>colMapList){
    	StringBuffer sbn = new StringBuffer();
    	StringBuffer sbv = new StringBuffer();
    	for (TargetColMapPO po : colMapList) {
    		sbn.append(po.getTargetColumn()+",");
    		sbv.append(" ? ,");
		}
    	// 先判断这个目标表是否可以进行数据恢复
    	if(targetTablePO.get().getCanDataRecovery() == 1){//可以进行数据恢复
    		sbn.append(" dm_task_id, dm_detail_id ,");
    		sbv.append(" ?, ? ,");
    	}
    	String strn = sbn.toString().substring(0, sbn.length()-1);
    	String strv = sbv.toString().substring(0, sbv.length()-1);
    	//拼接insert语句
    	StringBuffer sb = new StringBuffer("insert into ");
    	sb.append(targetTableName);
    	sb.append(" ( ");
    	sb.append(strn);
    	sb.append(" ) values (");
    	sb.append(strv);
    	sb.append(")");
    	LOGGER.info("{0} insert sql:{1}",Thread.currentThread(),sb.toString());
    	return sb.toString();
    }
    
    /**
     * 根据源对象拼接表名
     * @param baseName 基表名
     * @param subFlag 分表标识
     * @param dataId 对象标识
     * @param sourceDataPo 源对象
     * @param subMap 源对象值
     * @return
     */
    private String getTableName(String baseName,String subFlag,int dataId,int dataType,DmSrcTablePO sourceDataPo,Map<String ,Object> subMap){
		StringBuffer tableName = new StringBuffer();
		tableName.append(baseName);
		if(DMSConstant.SUB_FLAG_T.equals(subFlag)){
			//根据目标标识查找分表定义
			SubTableDefineArg subTableDefineArg = new SubTableDefineArg();
			SubTableDefineArg.SubTableDefineCriteria subTableDefineCriteria = subTableDefineArg.createCriteria();
			subTableDefineCriteria.andSourceDataIdEqualTo(dataId);
			subTableDefineCriteria.andDataTypeEqualTo(dataType);
			subTableDefineCriteria.andStateEqualTo(DMSConstant.Status.SUCCESS.toString());
			subTableDefineArg.setOrderByClause("priority");
			List<SubTableDefinePO> subTableDefineList = subTableDefineDao.selectByArg(subTableDefineArg);
			if(subTableDefineList.size()>1){
				for(int i=1;i<subTableDefineList.size();i++){
					SubTableDefinePO po = subTableDefineList.get(i);
					if(po.getSubDefineType()==DMSConstant.SUB_DEFINE_TYPE_FIXED){
						tableName.append("_"+po.getSubDefineValue());
					}else if(po.getSubDefineType()==DMSConstant.SUB_DEFINE_TYPE_CHANGE){
						Object subRefColValue =subMap.get(po.getSubRefCol());
						if(Utils.isEmpty(po.getSubDefineValue())){
							tableName.append("_"+String.valueOf(subRefColValue));
						}else{
							switch (Integer.parseInt(po.getSubDefineValue())) {
								case 2:
									tableName.append("_"+exchangeFunc.yearMonthExchange((Date)subRefColValue));
									break;
								case 3:
									tableName.append("_"+exchangeFunc.dateExchange((Date)subRefColValue));
									break;
								case 4:
									tableName.append("_"+exchangeFunc.billingCycleExchange((Date)subRefColValue));
									break;
								default:
									tableName.append("_"+String.valueOf(subRefColValue));
									break;
							}
						}
					}
				}
			}
		}
		return tableName.toString();
	}
    
    /**
     * 根据源表对象标识查找源表的查询字段
     * @param sourceId 源表对象标识
     * @param map 保存源字段在数据库中的类型
     * @return 源表的查询字段
     */
    private String getSearchColsBySourceId(int sourceId,Map<String,Integer>map){
    	StringBuffer sourceCol = new StringBuffer();
		TargetColMapArg targetColMapArg = new TargetColMapArg();
		TargetColMapArg.TargetColMapCriteria targetColMapCriteria = targetColMapArg.createCriteria();
		targetColMapCriteria.andSourceDataIdEqualTo(sourceId);
		targetColMapCriteria.andStateEqualTo(DMSConstant.Status.SUCCESS.toString());
		targetColMapCriteria.andSourceColFlagEqualTo("T");
		targetColMapArg.setOrderByClause("SEQ_NBR");
		List<TargetColMapPO> targetColMapList = targetColMapDao.selectByArg(targetColMapArg);
		if(targetColMapList.size()>0){
			for (TargetColMapPO targetColMapPO : targetColMapList) {
				sourceCol.append("," + targetColMapPO.getSourceColumn());
				if(Utils.isEmpty(map.get(targetColMapPO.getSourceColumn()))){
					map.put(targetColMapPO.getSourceColumn(),targetColMapPO.getSourceColumnType());
				}
			}
			return sourceCol.toString();
		}else{
			return "false";
		}
    }

    /**
     * 
     * @param pstm 预编译对象
     * @param list 参数列表
     * @throws SQLException
     */
    private void pstSetObject(PreparedStatement pstm,String columns,String colNames,Map<String,Integer>dataTypeMap,Map<String,Object>srcDataMap,String dbLinkCategory) throws SQLException   {  
    	String str[] = columns.split(",");
    	String strName[] = colNames.split(",");
    	Object value ;
        for(int i = 0; i < str.length; i++){  
        	if(Utils.isEmpty(srcDataMap)){
        		value = str[i];
        	}else{
        		value = srcDataMap.get(str[i]);
        	}
        	if(DMSConstant.DB_LINK_CATEGORY_ORACLE.equals(dbLinkCategory.toUpperCase())){
        		if(dataTypeMap.get(strName[i])==Types.DATE || dataTypeMap.get(strName[i]) == Types.TIMESTAMP
        				|| dataTypeMap.get(strName[i])== Types.TIME){
        			pstm.setTimestamp(i+1, Timestamp.valueOf(value.toString()));
        		}else{
        			pstm.setObject(i + 1, value);  
        		}
        	}else{
        		pstm.setObject(i + 1, value);  
        	}
        } 
    }
    
    /**
     * 主表拼接insert语句
     * @param conn
     * @param targetName
     * @param backUpMap
     * @param colMapList
     * @param sourceResultMap
     * @param targetBase
     * @return
     * @throws SQLException
     */
    private PreparedStatement InsertBatch(Connection conn,String targetName,Map<String,List<Long>>backUpMap,
    		List<TargetColMapPO>colMapList,Map<Long,Map<String,Object>>sourceResultMap) throws SQLException{
		List<Long> detailIdList = backUpMap.get(targetName);
		TargetColMapPO temptargetPo;
		Map<String,Object> sourceDataMap;
		String insertSql = insertSql(targetName,colMapList);
    	PreparedStatement pstmt = conn.prepareStatement(insertSql); 
		for (Long id : detailIdList) {
			sourceDataMap = sourceResultMap.get(id);
			if(!Utils.isEmpty(sourceDataMap)){
				for(int i=0;i<colMapList.size();i++){
					temptargetPo = colMapList.get(i);
					if(Utils.isEmpty(temptargetPo.getTargetColVal())){
						/*if(Utils.isEmpty(sourceDataMap.get(temptargetPo.getSourceColumn()))){
							pstmt.setNull(i+1, temptargetPo.getSourceColumnType());
						}else{*/
							pstmt.setObject(i+1,sourceDataMap.get(temptargetPo.getSourceColumn()));
					//	}
					}else{
						if(temptargetPo.getSourceColumnType() ==Types.DATE
								||temptargetPo.getSourceColumnType() ==Types.TIMESTAMP){
							//pstmt.setTimestamp(i+1, Timestamp.valueOf(temptargetPo.getTargetColVal()));
							pstmt.setTimestamp(i+1, new Timestamp(System.currentTimeMillis()));
							
						}else{
							pstmt.setObject(i+1,temptargetPo.getTargetColVal());
						}
					}
				}
				// 先判断这个目标表是否可以进行数据恢复
				if (targetTablePO.get().getCanDataRecovery() == 1) {// 可以进行数据恢复
					// 将当前任务ID写入目标表的DM_TASK_ID字段，将每个任务结果明细ID写入目标表的DM_DETAIL_ID字段
					pstmt.setObject(colMapList.size() + 1, dtsPO.getDmTaskId());
					pstmt.setObject(colMapList.size() + 2, id);
				}
				//加入批处理
				pstmt.addBatch();
			}
        }
		return pstmt;
    }
    
    /**
     *从表拼接insert语句
     * @param conn
     * @param targetName
     * @param backUpMap
     * @param colMapList
     * @param sourceResultMap
     * @return
     * @throws SQLException
     */
    private PreparedStatement InsertBatchForSub(Connection conn,String targetName,Map<String,List<Long>>backUpMap,
    		List<TargetColMapPO>colMapList,Map<Long,List<Map<String,Object>>>sourceResultMap) throws SQLException{
		List<Long> detailIdList = backUpMap.get(targetName);
		TargetColMapPO temptargetPo;
		//Map<String,Object> sourceDataMap;
		List<Map<String,Object>> tempList;
    	PreparedStatement pstmt = conn.prepareStatement(insertSql(targetName,colMapList)); 
		for (Long id : detailIdList) {
			tempList = sourceResultMap.get(id);
			for(Map<String,Object> map:tempList){
				if(!Utils.isEmpty(map)){
					for(int i=0;i<colMapList.size();i++){
						temptargetPo = colMapList.get(i);
						if(Utils.isEmpty(temptargetPo.getTargetColVal())){
							pstmt.setObject(i+1,map.get(temptargetPo.getSourceColumn()));
						}else{
							if(temptargetPo.getSourceColumnType() ==Types.DATE
									||temptargetPo.getSourceColumnType() ==Types.TIMESTAMP){
								pstmt.setTimestamp(i+1, new Timestamp(System.currentTimeMillis()));
							}else{
								pstmt.setObject(i+1,temptargetPo.getTargetColVal());
							}
						}
					}
			}
				// 先判断这个目标表是否可以进行数据恢复
				if (targetTablePO.get().getCanDataRecovery() == 1) {// 可以进行数据恢复
					// 将当前任务ID写入目标表的DM_TASK_ID字段，将每个任务结果明细ID写入目标表的DM_DETAIL_ID字段
					pstmt.setObject(colMapList.size() + 1, dtsPO.getDmTaskId());
					pstmt.setObject(colMapList.size() + 2, id);
				}
				//加入批处理
				pstmt.addBatch();
			}
        }
		return pstmt;
    }
    
    /**
     * @param conn
     * @param targetName
     * @param backUpMap
     * @param colMapList
     * @param sourceResultMap
     * @param targetBase
     * @return
     * @throws SQLException
     */
    private PreparedStatement deleteBatch(Connection conn,String targetName,Map<String,List<Long>>backUpMap,
    		List<TargetColMapPO>colMapList,Map<Long,Map<String,Object>>sourceResultMap) throws SQLException{
		List<Long> detailIdList = backUpMap.get(targetName);
		TargetColMapPO temptargetPo;
		Map<String,Object> sourceDataMap;
		StringBuffer sb = new StringBuffer("delete from "+targetName+" where 1=1 ");
    	for (TargetColMapPO po : colMapList) {
    		sb.append(" and "+po.getTargetColumn()+" = ? ");
		}
    	PreparedStatement pstmt = conn.prepareStatement(sb.toString()); 
		for (Long id : detailIdList) {
			sourceDataMap = sourceResultMap.get(id);
			if(!Utils.isEmpty(sourceDataMap)){
				for(int i=0;i<colMapList.size();i++){
					temptargetPo = colMapList.get(i);
					if(Utils.isEmpty(temptargetPo.getTargetColVal())){
							pstmt.setObject(i+1,sourceDataMap.get(temptargetPo.getSourceColumn()));
					}else{
						if(temptargetPo.getSourceColumnType() ==Types.DATE
								||temptargetPo.getSourceColumnType() ==Types.TIMESTAMP){
							pstmt.setTimestamp(i+1, new Timestamp(System.currentTimeMillis()));
							
						}else{
							pstmt.setObject(i+1,temptargetPo.getTargetColVal());
						}
					}
				}
			}
            //加入批处理
            pstmt.addBatch();
        }
		return pstmt;
    }
    private PreparedStatement deleteBatchForSub(Connection conn,String targetName,Map<String,List<Long>>backUpMap,
    		List<TargetColMapPO>colMapList,Map<Long,List<Map<String,Object>>>sourceResultMap) throws SQLException{
		List<Long> detailIdList = backUpMap.get(targetName);
		TargetColMapPO temptargetPo;
		List<Map<String,Object>> sourceDataMapList;
		StringBuffer sb = new StringBuffer("delete from "+targetName+" where 1=1 ");
    	for (TargetColMapPO po : colMapList) {
    		sb.append(" and "+po.getTargetColumn()+" = ? ");
		}
    	PreparedStatement pstmt = conn.prepareStatement(sb.toString()); 
		for (Long id : detailIdList) {
			sourceDataMapList = sourceResultMap.get(id);
			for(Map<String,Object> map :sourceDataMapList){
				if(!Utils.isEmpty(map)){
					for(int i=0;i<colMapList.size();i++){
						temptargetPo = colMapList.get(i);
						if(Utils.isEmpty(temptargetPo.getTargetColVal())){
								pstmt.setObject(i+1,map.get(temptargetPo.getSourceColumn()));
						}else{
							if(temptargetPo.getSourceColumnType() ==Types.DATE
									||temptargetPo.getSourceColumnType() ==Types.TIMESTAMP){
								pstmt.setTimestamp(i+1, new Timestamp(System.currentTimeMillis()));
								
							}else{
								pstmt.setObject(i+1,temptargetPo.getTargetColVal());
							}
						}
					}
				}
			}
            //加入批处理
            pstmt.addBatch();
        }
		return pstmt;
    }
    
    private List<Long>getKeyList(List<DmPO>dmList){
    	List<Long> list = new ArrayList<Long>();
    	for (DmPO po : dmList) {
    		list.add(po.getDmDetailId());
		}
    	return list;
    }

}
